package com.atbeijing.D09;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import static org.apache.flink.table.api.Expressions.$;

/**
 * 追加流和回滚流
 * 回滚流有true和false,聚合的时候false表示数据已失效
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //配置表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();

        //获取表环境
        StreamTableEnvironment TEnv = StreamTableEnvironment.create(env, settings);

        //数据源
        DataStreamSource<SensorReading> stream = env.addSource(new SensorSource());

        //建table
        Table table = TEnv.fromDataStream(
                stream,
                $("id"),
                $("temperature").as("temp"),
                $("timestamp").as("ts")

        );

        //查询
        Table selectResult = table.select(
                $("id"),
                $("temp")
        );

        //追加流
//        TEnv.toAppendStream(selectResult, Row.class).print();

        //回滚流,有true
        TEnv.toRetractStream(selectResult,Row.class).print();

        env.execute();
    }
}
